package cn.ipanel.bigdata.dw.dim.phoenix

import cn.ipanel.bigdata.boot.source.genre.Phoenix
import cn.ipanel.bigdata.dw.dws.HBASE_NAMESPACE
import cn.ipanel.bigdata.utils.Dictionary.F_DATE
import org.apache.spark.sql.{Dataset, Row}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_panyu   番禺大数据，区域数据
 * @date 2023/11/14 16:15
 * @description:
 */
protected[phoenix] object IOTArea extends Phoenix(HBASE_NAMESPACE, "t_iot_area") with Area {

  override def getTBColumns: Seq[String] = Seq(
    F_DATE, F_ID, F_COUNTY_ID, F_COUNTY_NAME, F_TOWN_ID, F_TOWN_NAME,
    F_NEIGHBORHOOD_ID, F_NEIGHBORHOOD_NAME, F_LEVEL
  )

  override def buildTable: String = {
    s"""
       |CREATE TABLE IF NOT EXISTS $getDBName.$getTBName(
       |$F_DATE INTEGER NOT NULL
       |, $F_ID BIGINT NOT NULL
       |, $F_COUNTY_ID BIGINT NOT NULL
       |, $F_COUNTY_NAME VARCHAR
       |, $F_TOWN_ID BIGINT NOT NULL
       |, $F_TOWN_NAME VARCHAR
       |, $F_NEIGHBORHOOD_ID BIGINT NOT NULL
       |, $F_NEIGHBORHOOD_NAME VARCHAR
       |, $F_LEVEL TINYINT
       |CONSTRAINT PK PRIMARY KEY($F_DATE,$F_ID)
       |) SALT_BUCKETS = $saltBuckets;
       |""".stripMargin
  }

  override def emptyTable: Dataset[_] = {
    import IMPLICITS._
    spark.createDataset(spark.sparkContext.emptyRDD[IOTAreaTable])
  }

  case class IOTAreaTable(var f_date: Int
                          , var f_id: Long
                          , var f_county_id: Long
                          , var f_county_name: String
                          , var f_town_id: Long
                          , var f_town_name: String
                          , var f_neighborhood_id: Long
                          , var f_neighborhood_name: String
                          , var f_last_p_id_array: scala.collection.mutable.WrappedArray[String]
                          , var f_last_s_name_array: scala.collection.mutable.WrappedArray[String]
                         )

  object IOTAreaTable {
    def apply(row: Row): IOTAreaTable = {
      IOTAreaTable(row.getAs[Int](F_DATE)
        , row.getAs[Long](F_ID)
        , row.getAs[Long](F_COUNTY_ID)
        , row.getAs[String](F_COUNTY_NAME)
        , row.getAs[Long](F_TOWN_ID)
        , row.getAs[String](F_TOWN_NAME)
        , row.getAs[Long](F_NEIGHBORHOOD_ID)
        , row.getAs[String](F_NEIGHBORHOOD_NAME)
        , row.getAs[scala.collection.mutable.WrappedArray[String]](F_LAST_P_ID_ARRAY)
        , row.getAs[scala.collection.mutable.WrappedArray[String]](F_LAST_S_NAME_ARRAY)
      )
    }
  }
}